package com.pingan.haofang.searchcloud.collection.service.impl;

import com.pingan.haofang.framework.common.functional.lang.data.Try;
import com.pingan.haofang.searchcloud.api.StorageCluster;
import com.pingan.haofang.searchcloud.api.facade.CollectionManageFacade;
import com.pingan.haofang.searchcloud.api.matedata.CollectionMetaData;
import com.pingan.haofang.searchcloud.api.route.IndexRoute;
import com.pingan.haofang.searchcloud.api.route.IndexRoute.Endpoint;
import com.pingan.haofang.searchcloud.api.route.IndexRouteRegisterFactory;
import com.pingan.haofang.searchcloud.collection.dao.CollectionDao;
import com.pingan.haofang.searchcloud.collection.domain.Collection;
import com.pingan.haofang.searchcloud.collection.service.CollectionService;
import com.pingan.haofang.searchcloud.common.constants.Status;
import com.pingan.haofang.searchcloud.common.exception.FieldException;
import com.pingan.haofang.searchcloud.common.exception.NotFoundException;
import com.pingan.haofang.searchcloud.index.dao.IndexDao;
import com.pingan.haofang.searchcloud.index.domain.Index;
import com.pingan.haofang.searchcloud.metadata.dao.MetadataDao;
import com.pingan.haofang.searchcloud.metadata.domain.Metadata;
import com.pingan.haofang.searchcloud.solr.collection.impl.CollectionCreateException;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;

/**
 * 索引Collection信息
 *
 * @author baojie371
 * @date 17/5/11
 * @since 1.0.0
 */
@Service
@ManagedResource
public class CollectionServiceImpl implements CollectionService {

    private static Logger LOG = LoggerFactory.getLogger(CollectionServiceImpl.class);

    private static final String SPLITOR = ",";

    @Autowired
    private CollectionDao collectionDao;

    @Autowired
    private CollectionManageFacade collectionManageFacade;

    @Autowired
    private MetadataDao metadataDao;

    @Autowired
    private IndexDao indexDao;


    @Override
    public Optional<Collection> getCollection(Index index) {
        List<Collection> collections = collectionDao.findByIndexIdAndStatus(index.getId(), Status.NORMAL.getValue());
        if (CollectionUtils.isEmpty(collections)) {
            return Optional.empty();
        } else {
            return Optional.of(collections.get(0));
        }
    }

    @Override
    @Transactional(rollbackFor = Exception.class )
    public Collection getOrCreateCollection(Index index, Metadata metadata, String collectionName) {
        List<Collection> collections = collectionDao.findByMetaIdAndCollectionName(metadata.getId(), collectionName);
        if (CollectionUtils.isEmpty(collections)) {

            // 新建Collection后返回
            Collection collection = new Collection();
            collection.setIndexId(index.getId());
            collection.setMetaId(metadata.getId());
            collection.setStatus(Status.STOP.getValue());
            collection.setCollectionName(collectionName);

            boolean created = Try.of(() -> collectionManageFacade.create(
                    new StorageCluster(metadata.getHost(), index.getEngineType()),
                    createCollectionMetaData(metadata, collection))).onFailure(e -> {
                if (e instanceof CollectionCreateException) {
                    throw new FieldException("index.replication.refactor.too.max.incluster", ArrayUtils
                            .add(new Object[0], ((CollectionCreateException) e).getMaxReplicationFactor()));
                }
            }).orElseThrow(e -> new RuntimeException("Create Failed", e));

            if (created) {

                collectionDao.saveAndFlush(collection);

                return collection;
            } else {
                throw new RuntimeException("Create Failed");
            }

        } else {
            return collections.get(0);
        }
    }

    @Override
    @Transactional
    public void registerCollection(Index index, String collectionName) {
        List<Collection> collections = collectionDao.findByIndexIdAndCollectionName(index.getId(), collectionName);
        if (CollectionUtils.isEmpty(collections)) {
            throw new NotFoundException("Collection Not Found");
        }

        Collection collection = collections.get(0);

        if (CollectionUtils.isEmpty(collections)) {
            throw new NotFoundException("Collection Not Found");
        }

        // 修改旧的collection状态
        List<Collection> oldCollections = collectionDao.findByIndexIdAndStatus(index.getId(), Status.NORMAL.getValue());
        if (CollectionUtils.isNotEmpty(oldCollections)) {
            oldCollections.forEach(oldCollection -> oldCollection.setStatus(Status.STOP.getValue()));
            collectionDao.save(oldCollections);
            collectionDao.flush();
        }

        collection.setStatus(Status.NORMAL.getValue());
        collectionDao.saveAndFlush(collection);

        registerCollection2zk(index, collection, oldCollections);
    }

    private void registerCollection2zk(Index index, Collection normalCollection, List<Collection> historyCollections) {

        IndexRoute route = new IndexRoute();

        route.setEngineType(index.getEngineType());
        route.setCollectionRouteKey(index.getIndexName());
        route.setCollectionName(normalCollection.getCollectionName());
        route.setEffectYYMMddHHmmTime(DateFormatUtils.format(new Date(), "yyyy-MM-dd'T'HH:mm:ss"));

        Metadata meta = metadataDao.getOne(normalCollection.getMetaId());
        route.setEngineHost(meta.getHost());

        if (CollectionUtils.isNotEmpty(historyCollections)) {
            Set<Endpoint> endpointSet = new HashSet<Endpoint>(historyCollections.size());
            for (Collection c : historyCollections) {
                Metadata metaTemp = metadataDao.getOne(c.getMetaId());
                Endpoint routeEndpoint = new Endpoint().buildName(c.getCollectionName()).buildEffectTime(new Date())
                        .buildEngineType(index.getEngineType()).buildEngineHost(metaTemp.getHost());
                if (endpointSet.contains(routeEndpoint)) {
                    endpointSet.remove(routeEndpoint);
                }
                endpointSet.add(routeEndpoint);
            }
            route.setEndpoints(new ArrayList<Endpoint>(endpointSet));
            Collections.sort(route.getEndpoints());
        }

        IndexRouteRegisterFactory.get().register(route);
    }

    @Override
    public List<Long> clearOldCollectionsOfStorage(long indexId, int engineType, int topN) {

        List<Collection> collections = collectionDao.findByIndexIdAndStatusOrderByIdDesc(indexId,
                Status.STOP.getValue());
        int n = 0;
        List<Long> collectionIds = new ArrayList<>(collections.size());
        for (Collection collection : collections) {
            n++;
            if (n > topN) {
                try {
                    Metadata metadata = metadataDao.findOne(collection.getMetaId());
                    if (collectionManageFacade.delete(new StorageCluster(metadata.getHost(), engineType)
                            , collection.getCollectionName())) {

                        collectionIds.add(collection.getId());
                    } else {
                        LOG.error("Collection Delete Error");
                        collectionIds.add(collection.getId());
                    }
                } catch (Exception e) {
                    LOG.error("Collection Delete Exception", e);
                    collectionIds.add(collection.getId());
                }
            }
        }
        return collectionIds;
    }

    @Override
    @Transactional
    public void clearOldCollectionsOfDb(List<Long> collectionIds) {
        List<Collection> collections = collectionDao.findAll(collectionIds);
        collections.forEach(collection -> collection.setStatus(Status.DELETE.getValue()));
        collectionDao.save(collections);
    }



    /**
     * 获取未使用collection名
     *
     * @param engineType
     * @param host
     * @return
     */
    @ManagedOperation
    public List<String> getUnusedCollectionNames(int engineType, String host) throws Exception {
        StorageCluster storageCluster = new StorageCluster(host, engineType);
        List<String> collectionNames = collectionManageFacade.listAllCollectionNames(storageCluster);

        List<String> unUsedCollectionNames = new ArrayList<>();
        for (String name : collectionNames) {
            //判断是否满足未使用条件
            if(isUnUsedCollection(name)){
                unUsedCollectionNames.add(name);
            }
        }

        LOG.info("unused collections size {}: {}", CollectionUtils.size(unUsedCollectionNames), unUsedCollectionNames);
        return unUsedCollectionNames;
    }

    /**
     * 判断collection是否满足未使用条件
     * @param name
     * @return
     */
    private Boolean isUnUsedCollection(String name) {
        //index表需要有记录
        if (getIndexNameOfCollection(name) == null) {
            return false;
        }

        List<Collection> collections = collectionDao.findByCollectionName(name);

        //数据库无collection记录
        if (CollectionUtils.isEmpty(collections)) {
            return true;
        }
        //数据库collection记录状态是删除
        Integer status = collections.get(0).getStatus();
        if (status == Status.DELETE.getValue()) {
            return true;
        }
        return false;

    }


    /**
     * 查询collectionName对应的indexName(数据库存在)
     *
     * @param collectionName
     * @return
     */
    @Override
    public Index getIndexNameOfCollection(String collectionName) {
        String[] nameArr = collectionName.split("_");
        if (nameArr.length < 3) {
            LOG.info("collection {} is not searchcloud cllection", collectionName);
            return null;
        }
        String[] expectedIndexNameArr = new String[nameArr.length - 2];
        System.arraycopy(nameArr, 0, expectedIndexNameArr, 0, expectedIndexNameArr.length);

        String expectedIndexName = StringUtils.join(expectedIndexNameArr, "_");
        List<Index> indexList = indexDao.findByIndexName(expectedIndexName);
        if (CollectionUtils.isNotEmpty(indexList)) {
            return indexList.get(0);
        } else {
            LOG.info("collection {} is not searchcloud cllection", collectionName);
        }
        return null;
    }

    /**
     * 获取未使用collection名
     *
     * @param engineType
     * @param host
     * @return
     */
    @ManagedOperation
    public List<String> clearUnusedCollections(int engineType, String host) throws Exception {
        List<String> collectionNames = getUnusedCollectionNames(engineType, host);
        return clearOldCollectionsOfStorage(engineType, host, StringUtils.join(collectionNames, SPLITOR));
    }

    /**
     * 清理Solr里旧的Collections，无数据库关联
     *
     * @param engineType
     * @param host
     * @param collectionNamesStr 逗号分隔
     * @return
     */
    @ManagedOperation
    public List<String> clearOldCollectionsOfStorage(int engineType, String host, String collectionNamesStr) {
        List<String> deleted = new ArrayList<>();

        String[] collectionNames = collectionNamesStr.split(SPLITOR);
        for (String name : collectionNames) {
            try {

                if (collectionManageFacade.delete(new StorageCluster(host, engineType), StringUtils.trim(name))) {
                    LOG.info("collection of engineType {} : {} deleted ", engineType, name);
                    deleted.add(name);
                }
            } catch (Exception e) {
                LOG.error("Collection Delete Exception", e);
            }
        }
        LOG.info("unused collection of engineType {} : {} deleted ", engineType, collectionNames);
        return deleted;
    }

    private CollectionMetaData createCollectionMetaData(Metadata metadata, Collection collection) {
        CollectionMetaData collectionMetaData = new CollectionMetaData();
        collectionMetaData.setName(collection.getCollectionName());
        collectionMetaData.setConfigName(formatConfigNameInZk(metadata));
        collectionMetaData.setAutoAddReplicas(metadata.getAutoAddReplicas());
        collectionMetaData.setMaxShardsPerNode(metadata.getMaxShardsPerNode());
        collectionMetaData.setReplicationFactor(metadata.getReplicationFactor());
        collectionMetaData.setShardCount(metadata.getShardCount());
        CollectionMetaData.Router router = new CollectionMetaData.Router();
        router.setName(metadata.getRouterName());
        collectionMetaData.setRouter(router);
        return collectionMetaData;
    }

    private String formatConfigNameInZk(Metadata metadata) {
        return String.format("%s_%d", metadata.getConfigName(), metadata.getConfigVersion());
    }


}
